package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/3/14
 */
public class App08_DwdMhsPreoperativeVisit extends BaseSqlApp {
    public static void main(String[] args) {
        new App08_DwdMhsPreoperativeVisit().init(
                5008,
                "App8_DwdMhsPreoperativeVisit",
                2
        );
    }

    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        // 创建Kafka映射表
        tableEnv.executeSql("create table target (" +
                "RECORD_ID                      String,"+
                "MED_ORG_NAME                   String,"+
                "MED_ORG_CODE                   String,"+
                "HOSPITAL_NAME                  String,"+
                "HOSPITAL_CODE                  String,"+
                "AGE_UNIT                       String,"+
                "AGE                            String,"+
                "SEX_NAME                       String,"+
                "SEX_CODE                       String,"+
                "ID_CARD_NUMBER                 String,"+
                "PATIENT_VISIT_CATEGORY_CODE    String,"+
                "PATIENT_VISIT_CATEGORY         String,"+
                "INPATIENT_VISIT_FLOW_ID        String,"+
                "INPATIENT_NO                   String,"+
                "OUTPATIENT_VISIT_FLOW_ID       String,"+
                "OUTPATIENT_NO                  String,"+
                "PATIENT_ID                     String,"+
                "PATIENT_NAME                   String,"+
                "DOMAIN_ID                      String,"+
                "BED_NO                         String,"+
                "BED_NAME                       String,"+
                "WARD_ID                        String,"+
                "WARD_NAME                      String,"+
                "VISIT_DATE                     String,"+
                "VISIT_TIMES                    String,"+
                "APPY_SHEET_NO                  String,"+
                "AUTHOR_CODE                    String,"+
                "AUTHOR_NAME                    String,"+
                "CUSTODIAN_CODE                 String,"+
                "CUSTODIAN_NAME                 String,"+
                "ANESTH_SIGNDATE                String,"+
                "ANESTH_CODE                    String,"+
                "ANES_DOCTOR_NAME               String,"+
                "ADMIT_DATETIME                 String,"+
                "DEPT_ID                        String,"+
                "DEPT_NAME                      String,"+
                "PREOPERATIVE_COMPLICATION      String,"+
                "BRIEF_MEDICAL_HISTORY          String,"+
                "ALLERGY_HISTORY                String,"+
                "PATIENT_WEIGHT                 String,"+
                "MENTAL_STATE_FLAG              String,"+
                "ABO_BLOOD_CODE                 String,"+
                "ABO_BLOOD_NAME                 String,"+
                "RH_BLOOD_CODE                  String,"+
                "RH_BLOOD_NAME                  String,"+
                "OPERATE_ROOM_NO                String,"+
                "PRE_OPER_DIAG_ICD_CODE         String,"+
                "PRE_OPER_DIAG_ICD_NAME         String,"+
                "CONSIDERATIONS                 String,"+
                "CARDIOGRAM_EXAM_RESULT         String,"+
                "CHEST_RAY_EXAM_RESULT          String,"+
                "CT_EXAM_RESULT                 String,"+
                "B_ULTRAPHONIC_EXAM_RESULT      String,"+
                "MRI_EXAM_RESULT                String,"+
                "LUNG_FUNCTION_EXAM_RESULT      String,"+
                "BLOOD_ROUTINE_EXAM_RESULT      String,"+
                "URINE_ROUTINE_EXAM_RESUTL      String,"+
                "COAGULATION_EXAM_RESULT        String,"+
                "LIVER_FUNCTION_EXAM_RESULT     String,"+
                "BLOOD_GAS_ANALYSIS_RESULT      String,"+
                "GENERAL_EXAM_RESULT            String,"+
                "HEART_AUSCULTATE_RESULT        String,"+
                "LUNG_AUSCULTATE_RESULT         String,"+
                "ROOM_NO                        String,"+
                "ROOM_NAME                      String,"+
                "LIMBS_EXAM_RESULT              String,"+
                "SPINE_EXAM_RESULT              String,"+
                "STOMACH_EXAM_RESULT            String,"+
                "TRACHEA_EXAM_RESULT            String,"+
                "TOOTH_EXAM_RESULT              String,"+
                "ANES_INDICATION                String,"+
                "ANES_DOCTOR_ID                 String,"+
                "ANES_DOCTOR_SIGN_NAME          String,"+
                "ELEC_SIGN_DATETIME             String,"+
                "INTEND_ANES_METHOD_CODE        String,"+
                "INTEND_ANES_METHOD_NAME        String,"+
                "OPER_CODE                      String,"+
                "OPER_NAME                      String," +
                "VISIT_ID                       String,"+
                "primary key (RECORD_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_PREOPERATIVE_VISIT));
        // 导入数据
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        //1 设置数据的TTL（暂时10min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        //2 join操作
        Table joinResult = tableEnv.sqlQuery("select " +
                "s.unique_id	RECORD_ID                                                                            ,"+
                "hospital_name	MED_ORG_NAME                                                                         ,"+
                "hospital_code	MED_ORG_CODE                                                                         ,"+
                "hospital_name	HOSPITAL_NAME                                                                        ,"+
                "hospital_code	HOSPITAL_CODE                                                                        ,"+
                "age_unit	AGE_UNIT                                                                                 ,"+
                "age	AGE                                                                                          ,"+
                "sex	SEX_NAME                                                                                     ,"+
                "sex_code	SEX_CODE                                                                                 ,"+
                "identify_no	ID_CARD_NUMBER                                                                       ,"+
                "patient_typecode	PATIENT_VISIT_CATEGORY_CODE                                                      ,"+
                "patient_type	PATIENT_VISIT_CATEGORY                                                               ,"+
                "case when patient_typecode ='3' then visit_id else '' end INPATIENT_VISIT_FLOW_ID                   ,"+
                "inpatient	INPATIENT_NO                                                                             ,"+
                "case when patient_typecode <> '3' then visit_id else '' end	OUTPATIENT_VISIT_FLOW_ID             ,"+
                "out_patient_id	OUTPATIENT_NO                                                                        ,"+
                "patient_id	PATIENT_ID                                                                               ,"+
                "patient_name	PATIENT_NAME                                                                         ,"+
                "s.domain_id	DOMAIN_ID                                                                            ,"+
                "bed_no	BED_NO                                                                                       ,"+
                "bed_name	BED_NAME                                                                                 ,"+
                "wards_id	WARD_ID                                                                                  ,"+
                "wards_name	WARD_NAME                                                                                ,"+
                "visit_date	VISIT_DATE                                                                               ,"+
                "visit_times	VISIT_TIMES                                                                          ,"+
                "appy_id	APPY_SHEET_NO                                                                            ,"+
                "author_code	AUTHOR_CODE                                                                          ,"+
                "author_name	AUTHOR_NAME                                                                          ,"+
                "custodian_code	CUSTODIAN_CODE                                                                       ,"+
                "custodian_name	CUSTODIAN_NAME                                                                       ,"+
                "anesth_signdate	ANESTH_SIGNDATE                                                                  ,"+
                "anesth_code	ANESTH_CODE                                                                          ,"+
                "anesth_name	ANES_DOCTOR_NAME                                                                     ,"+
                "admission_date	ADMIT_DATETIME                                                                       ,"+
                "dept_code	DEPT_ID                                                                                  ,"+
                "dept_name	DEPT_NAME                                                                                ,"+
                "preoperative_complication	PREOPERATIVE_COMPLICATION                                                ,"+
                "brief_medical_history	BRIEF_MEDICAL_HISTORY                                                        ,"+
                "allergy_history	ALLERGY_HISTORY                                                                  ,"+
                "weight	PATIENT_WEIGHT                                                                               ,"+
                "mental_state_mark	MENTAL_STATE_FLAG                                                                ,"+
                "abo_code	ABO_BLOOD_CODE                                                                           ,"+
                "abo	ABO_BLOOD_NAME                                                                               ,"+
                "rh_code	RH_BLOOD_CODE                                                                            ,"+
                "rh	RH_BLOOD_NAME	                                                                                 ,"+
                "oper_room	OPERATE_ROOM_NO                                                                          ,"+
                "case when diag_type ='术前诊断' then diag_code else '' end	PRE_OPER_DIAG_ICD_CODE                   ,"+
                "case when diag_type ='术前诊断' then diag_name else '' end	PRE_OPER_DIAG_ICD_NAME                   ,"+
                "needing_attention	CONSIDERATIONS                                                                   ,"+
                "ecg_result	CARDIOGRAM_EXAM_RESULT                                                                   ,"+
                "chest_x_findings	CHEST_RAY_EXAM_RESULT                                                            ,"+
                "ct_result	CT_EXAM_RESULT                                                                           ,"+
                "bu_result	B_ULTRAPHONIC_EXAM_RESULT                                                                ,"+
                "mri_result	MRI_EXAM_RESULT                                                                          ,"+
                "pulmonary_exam	LUNG_FUNCTION_EXAM_RESULT                                                            ,"+
                "blood_exam	BLOOD_ROUTINE_EXAM_RESULT                                                                ,"+
                "routine_urine_exam	URINE_ROUTINE_EXAM_RESUTL                                                        ,"+
                "coagulation_test	COAGULATION_EXAM_RESULT                                                          ,"+
                "liver_function_exam	LIVER_FUNCTION_EXAM_RESULT                                                   ,"+
                "blood_gas_analysis	BLOOD_GAS_ANALYSIS_RESULT                                                        ,"+
                "general_exam	GENERAL_EXAM_RESULT                                                                  ,"+
                "auscultation_result	HEART_AUSCULTATE_RESULT                                                      ,"+
                "pulmonary_auscultation	LUNG_AUSCULTATE_RESULT                                                       ,"+
                "ward_id	ROOM_NO                                                                                  ,"+
                "ward_name	ROOM_NAME                                                                                ,"+
                "limb_exam	LIMBS_EXAM_RESULT                                                                        ,"+
                "spinal_exam	SPINE_EXAM_RESULT                                                                    ,"+
                "celiac_exam	STOMACH_EXAM_RESULT                                                                  ,"+
                "tracheal_exam	TRACHEA_EXAM_RESULT                                                                  ,"+
                "dental_exam	TOOTH_EXAM_RESULT                                                                    ,"+
                "appropriate_anes	ANES_INDICATION                                                                  ,"+
                "anesth_code	ANES_DOCTOR_ID                                                                       ,"+
                "anesth_name	ANES_DOCTOR_SIGN_NAME                                                                ,"+
                "effective_time	ELEC_SIGN_DATETIME                                                                   ,"+
                "g.anes_method_code	INTEND_ANES_METHOD_CODE                                                          ,"+
                "g.anes_method_name	INTEND_ANES_METHOD_NAME                                                          ,"+
                "oper_code	OPER_CODE                                                                                ,"+
                "oper_name	OPER_NAME                                                                                ," +
                "visit_id   VISIT_ID "+
                "from hdsd00_06_05 s  "+
                "left join 	tdy_list_diag f	on f.unique_id=s.unique_id and f.xds_type=s.xds_type		"+
                "left join 	tdy_list_qanes g	on g.unique_id=s.unique_id and g.xds_type=s.xds_type	"+
                "left join 	tdy_list_qoper a	on a.unique_id=s.unique_id and a.xds_type=s.xds_type	"+
                "");
        tableEnv.createTemporaryView("join_result",joinResult);
    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table hdsd00_06_05 (" +
                "pk                         String,"+
                "upload_time                String,"+
                "status                     String,"+
                "empi                       String,"+
                "encounter_id               String,"+
                "bed_name                   String,"+
                "ward_name                  String,"+
                "wards_id                   String,"+
                "visit_date                 String,"+
                "patient_id                 String,"+
                "patient_domain             String,"+
                "visit_domain               String,"+
                "visit_id                   String,"+
                "visit_times                String,"+
                "unique_id                  String,"+
                "xds_type                   String,"+
                "xds_name                   String,"+
                "effective_time             String,"+
                "xds_version                String,"+
                "domain_id                  String,"+
                "out_patient_id             String,"+
                "patient_type               String,"+
                "inpatient                  String,"+
                "patient_typecode           String,"+
                "appy_id                    String,"+
                "identify_no                String,"+
                "patient_name               String,"+
                "sex_code                   String,"+
                "sex                        String,"+
                "age                        String,"+
                "age_unit                   String,"+
                "write_time                 String,"+
                "author_code                String,"+
                "author_name                String,"+
                "custodian_code             String,"+
                "custodian_name             String,"+
                "anesth_signdate            String,"+
                "anesth_code                String,"+
                "anesth_name                String,"+
                "admission_date             String,"+
                "bed_no                     String,"+
                "ward_id                    String,"+
                "dept_code                  String,"+
                "dept_name                  String,"+
                "wards_name                 String,"+
                "hospital_code              String,"+
                "hospital_name              String,"+
                "preoperative_complication  String,"+
                "brief_medical_history      String,"+
                "allergy_history            String,"+
                "weight                     String,"+
                "general_exam               String,"+
                "mental_state_mark          String,"+
                "auscultation_result        String,"+
                "pulmonary_auscultation     String,"+
                "limb_exam                  String,"+
                "spinal_exam                String,"+
                "celiac_exam                String,"+
                "tracheal_exam              String,"+
                "dental_exam                String,"+
                "abo_code                   String,"+
                "abo                        String,"+
                "rh_code                    String,"+
                "rh                         String,"+
                "ecg_result                 String,"+
                "chest_x_findings           String,"+
                "ct_result                  String,"+
                "bu_result                  String,"+
                "mri_result                 String,"+
                "pulmonary_exam             String,"+
                "blood_exam                 String,"+
                "routine_urine_exam         String,"+
                "coagulation_test           String,"+
                "liver_function_exam        String,"+
                "blood_gas_analysis         String"+
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_06_05","App10_DwdMhsPreoperativeVisit"));

        tableEnv.executeSql("create table tdy_list_qoper (" +
                "pk               String,"+
                "unique_id        String,"+
                "domain_id        String,"+
                "xds_type         String,"+
                "xds_version      String,"+
                "anes_method_code String,"+
                "anes_method_name String,"+
                "contraindication String,"+
                "oper_code        String,"+
                "oper_date        String,"+
                "oper_method      String,"+
                "oper_name        String,"+
                "oper_room        String,"+
                "oper_surgical    String,"+
                "pre_preparation  String"+
                ")"+
                FlinkSourceUtil.getKafkaWith("tdy_list_qoper","App10_DwdMhsPreoperativeVisit"));

        super.readTdyListDiag(tableEnv,"App10_DwdMhsPreoperativeVisit");

        tableEnv.executeSql("create table tdy_list_qanes (" +
                "pk                 String,"+
                "unique_id          String,"+
                "domain_id          String,"+
                "xds_type           String,"+
                "xds_version        String,"+
                "anes_method_code   String,"+
                "anes_method_name   String,"+
                "appropriate_anes   String,"+
                "needing_attention  String,"+
                "pre_anes_order     String"+
                ")"+
                FlinkSourceUtil.getKafkaWith("tdy_list_qanes","App10_DwdMhsPreoperativeVisit"));
    }
}
